package com.fanxl.flink.datastream.windows;

import com.fanxl.flink.datastream.watermark.MyPeriodicGenerator;
import com.fanxl.flink.datastream.watermark.TuplePeriodicWatermark;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.runtime.operators.util.AssignerWithPeriodicWatermarksAdapter;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/10 11:02
 */
public class WindowApp {

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8098");
        configuration.setString("rest.flamegraph.enabled","true");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        test08(env);
        env.execute("WindowApp");
    }

    public static void test03(StreamExecutionEnvironment env) {
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of("pk", Integer.parseInt(value));
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new PKProcessWindowFunction())
                .print();
    }


    public static void test02(StreamExecutionEnvironment env) {
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        String[] splits = value.split(",");
                        return Tuple2.of(splits[0].trim(), Integer.parseInt(splits[1].trim()));
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {

                        System.out.println("value1 = [" + value1 + "], value2 = [" + value2 + "]");
                        return Tuple2.of(value1.f0, value1.f1+value2.f1);
                    }
                })
                .print();
    }


    public static void test01(StreamExecutionEnvironment env) {
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.parseInt(value);
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(0)
                .print();

        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.parseInt(value);
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(0)
                .print();


        // hadoop,1  spark,1
//        env.socketTextStream("localhost", 9527)
//                .map(new MapFunction<String, Tuple2<String,Integer>>() {
//                    @Override
//                    public Tuple2<String, Integer> map(String value) throws Exception {
//                        String[] splits = value.split(",");
//                        return Tuple2.of(splits[0].trim(), Integer.parseInt(splits[1].trim()));
//                    }
//                }).keyBy(x -> x.f0)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(1)
//                .print();
    }

    public static void test05(StreamExecutionEnvironment env) {

        env.socketTextStream("10.215.15.200", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of("pk", Integer.parseInt(value));
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner((event, timestamp) -> (long) event.f1)).keyBy(x -> x.f0)
                .window(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(1)))
                .process(new PKProcessWindowFunction())
                .print();
    }


    public static void test06(StreamExecutionEnvironment env) {

        env.socketTextStream("10.215.15.200", 9527)
                .map(new MapFunction<String, Tuple2<String,Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        return Tuple2.of("pk", Long.parseLong(value));
                    }
                }).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarksAdapter.Strategy<>(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(Tuple2<String, Long> stringLongTuple2) {
                return stringLongTuple2.f1;
            }
        }))
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(1)))
                .process(new ProcessAllWindowFunction<Tuple2<String, Long>, String, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
                        System.out.println("----process invoked...----");

                        long maxValue = Long.MIN_VALUE;
                        for (Tuple2<String, Long> element : iterable) {
                            maxValue = Math.max(element.f1, maxValue);
                        }

                        collector.collect("当前窗口的最大值是:" + maxValue);
                    }
                })
                .print();
    }

    public static void test07(StreamExecutionEnvironment env) {

        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        return Tuple2.of("pk", Long.parseLong(value));
                    }
                }).assignTimestampsAndWatermarks(new TuplePeriodicWatermark(false))
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(5)))
                .apply(new AllWindowFunction<Tuple2<String, Long>, String, TimeWindow>() {

                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
                        System.out.println("----process invoked...----");

                        long maxValue = Long.MIN_VALUE;
                        for (Tuple2<String, Long> element : iterable) {
                            maxValue = Math.max(element.f1, maxValue);
                        }

                        collector.collect("当前窗口的最大值是:" + maxValue);
                    }
                })

                .print();
    }

    public static void test08(StreamExecutionEnvironment env) {
        env.setParallelism(1);
        env.socketTextStream("10.215.15.200", 9527)
                .map(new MapFunction<String, Tuple2<String,Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        return Tuple2.of("pk", Long.parseLong(value));
                    }
                })
//                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarksAdapter.Strategy<>(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(2)) {
//                    @Override
//                    public long extractTimestamp(Tuple2<String, Long> element) {
//                        return element.f1;
//                    }
//                }))
//                .assignTimestampsAndWatermarks(new TuplePeriodicWatermark(false))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner((event, timestamp) ->  event.f1))
                .windowAll(SlidingEventTimeCustomWindows.of(Time.seconds(2)))
                .trigger( EventTimeCustomTrigger.create())
                .process(new ProcessAllWindowFunction<Tuple2<String, Long>, Object, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<Tuple2<String, Long>> elements, Collector<Object> out) throws Exception {
                        System.out.println("----process invoked...----"+ context.window().toString());

                        Iterator<Tuple2<String, Long>> iterator = elements.iterator();
                        Set<Tuple2<String, Long>> set = new HashSet<Tuple2<String, Long>>();
                        while(iterator.hasNext()) {
                            Tuple2<String, Long> next = iterator.next();
                            if(set.contains(next)){
                                //随便去重日志，完全相同的去掉
                                System.out.println("去重"+set);
                            }else{
                                set.add(next);
                            }
                        }
                        System.out.println("窗口内数据"+set);
                        out.collect("当前窗口的最大值是:" + set);
                    }
                }).forceNonParallel()
                .print();
    }
}
